package com.song.storm_demo.hbase_chap8.producer;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.List;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.apache.commons.io.IOUtils;

/**
 * 
 * @author root
 *
 */
public class DateProducer {
	
	
	private Producer<String, String> producer;
	
	private  void  prepare()
	{
		Properties props = new Properties();
		props.put("zk.connect", "dmg:2181");
		props.put("metadata.broker.list", "dmg:9092");
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		props.put("key.serializer.class", "kafka.serializer.StringEncoder");
		props.put("request.required.acks", "1");
		ProducerConfig config =new  ProducerConfig(props);
		producer = new Producer<String, String>(config);
	}
	
	
	public  void produceData()
	{
		
		try {
			List<String>  lines =IOUtils.readLines(new FileInputStream(new File("./data/data1")));
			
			for(String line :lines)
			{
				producer.send( new KeyedMessage<String, String>("song",line) );
			}
			
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		producer.close();
		
	}
	
	
	
	
	public static void main(String[] args)
	{
		DateProducer dateProducer = new DateProducer();
		dateProducer.prepare();
		dateProducer.produceData();
		
	}
	
	
	
}
